package cdrprocessing.bolt.rule;

import java.util.Map;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class RuleLoadBalancingBolt extends BaseRichBolt {

    private long tupleCount;

    private int numberOfPartitions;

    private OutputCollector output;

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declareStream("RuleStream0", new Fields("rule", "action"));
        declarer.declareStream("RuleStream1", new Fields("rule", "action"));
        declarer.declareStream("RuleStream2", new Fields("rule", "action"));
        declarer.declareStream("RuleStream3", new Fields("rule", "action"));
    }

    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        tupleCount = 0;
        numberOfPartitions = Integer.valueOf((String) stormConf.get("numberOfPartitions"));
        output = collector;
    }

    public void execute(Tuple input) {
        tupleCount++;
        output.emit("RuleStream" + tupleCount % numberOfPartitions, new Values(input.getValueByField("rule"), input.getStringByField("action")));
    }
}
